package com.kingsoft.dc.khaos.plugin.core.dts;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.kingsoft.dc.khaos.extender.meta.api.AuthRequest;
import com.kingsoft.dc.khaos.extender.meta.api.DmRequest;
import com.kingsoft.dc.khaos.extender.meta.model.MetaParams;
import com.kingsoft.dc.khaos.extender.meta.model.ds.DmDatasource;
import com.kingsoft.dc.khaos.extender.meta.model.ds.KyligenceConnect;
import com.kingsoft.dc.khaos.extender.meta.utils.DmMetaUtils;
import com.kingsoft.dc.khaos.plugin.core.constants.Contants;
import com.ksyun.kbdp.dts.common.utils.Props;
import com.ksyun.kbdp.dts.job.core.ProcessJob;
import com.ksyun.kbdp.dts.job.core.util.CommandUtil;
import org.apache.commons.io.FileUtils;
import org.apache.log4j.Logger;

import java.io.*;


/**
 * Created by jing on 19/8/2.
 */
public class KylinPluginSubmit extends ProcessJob {
    //    final String USER_SCRIPT_FILENAME = "user_script";//脚本名
//    final String MAIN_SCRIPT_FILENAME = "run";
    private Logger log = null;


    public KylinPluginSubmit(String jobid, Props sysProps, Props jobProps, Logger logger) {
        super(jobid, sysProps, jobProps, logger);
        log = logger;
    }

    @Override
    protected String getFileContent() {
        StringBuffer content = new StringBuffer();
        String pluginDir = sysProps.get("plugin.dir");
        log.info("=>pluginDir：" + pluginDir);
        String type = jobProps.get("type");
        String scriptFile = "";
        if (type.equalsIgnoreCase("kyligenceBatch")) {
            scriptFile = "build_cube.sh";
        } else if (type.equalsIgnoreCase("kyligenceStream")) {
            scriptFile = "build_cube_stream.sh";
        } else {
            log.error("=>unknown type >> " + type);
        }
        String code = readFile(pluginDir + "/" + scriptFile);
        content.append(code);
        return content.toString();
    }

    @Override
    protected String getCommand() {
        String executingDir = this.jobProps.getString("job.executing.dir");
        String pluginDir = sysProps.get("plugin.dir");
        File file = CommandUtil.command2File(executingDir, this.getFileContent(), this.getFileName());

        //判断插件类型，使用不同的脚本
        String type = jobProps.get("type");
        String scripParams = "";
        if (type.equalsIgnoreCase("kyligenceBatch")) {
            scripParams = getBatchScriptParams();
        } else if (type.equalsIgnoreCase("kyligenceStream")) {
            scripParams = getStreamScriptParams();
        } else {
            log.error("=>unknown type >> " + type);
        }
        copyKylinJqFile(pluginDir + "/jq", executingDir + "/jq");

        String command = "sh -x " + this.getFileName() + " " + scripParams;
        String json = jobProps.get("plugin_param");
        log.info("=>command: " + command);
        log.info("=>" + json);
        return command;
    }

    private void copyKylinJqFile(String sourcePath, String destPath) {
        try {
            File source = new File(sourcePath);
            File dest = new File(destPath);
            FileUtils.copyFile(source, dest);
            Process exec = Runtime.getRuntime().exec("chmod 777 " + destPath);
            int i = exec.waitFor();
            log.info("=>exec chmod jq result is " + i);
        } catch (Exception e) {
            log.error("=>copy jq file or chmod failed !", e);
        }
    }

    protected String readFile(String path) {
        String laststr = "";
        try {
            FileInputStream fileInputStream = new FileInputStream(path);
            InputStreamReader inputStreamReader = new InputStreamReader(fileInputStream, "UTF-8");
            BufferedReader reader = new BufferedReader(inputStreamReader);
            String tempString = "";
            while (tempString != null) {
                tempString = reader.readLine();
                if (tempString != null) {
                    laststr += tempString + "\n";
                }
            }
            reader.close();
        } catch (Exception e) {
//            e.printStackTrace();
            log.error("读取脚本内容失败！", e);
        }
        return laststr + "\n";
    }


    private JSONObject parsePluginJson(String str) {
        JSONObject pluginParam = JSON.parseObject(str);
        JSONObject config = JSON.parseObject(pluginParam.get("config").toString());
        return config;
    }

    private KyligenceConnect getDmDataSourceInfo() {
        int jobProjectId = jobProps.getInt("project.id");
        String runEnv = jobProps.get("job.run.env");
        //解析json
        JSONObject pluginJson = parsePluginJson(jobProps.get("plugin_param"));
        JSONObject datasourceJson = JSON.parseObject(pluginJson.getString("datasource"));
        JSONObject metaJson = JSON.parseObject(JSON.parseObject(datasourceJson.getString("meta")).getString("params"));
        log.info("=>projectId=" + jobProjectId);
        log.info("=>runEnv=" + runEnv);
        log.info("=>meta=" + metaJson);
//        int dsId = metaJson.getInteger("ds_id");
        String dsName = metaJson.getString("ds_name");

        DmRequest metaRequest = buildMetaRequest();
        MetaParams metaParams = new MetaParams(metaJson.toString());
        metaParams.setJobProjectId(jobProjectId);
//        metaParams.buildWriteDatasourceParams(runEnv, dsId);
        metaParams.buildWriteDatasourceParams(runEnv, dsName);

        metaRequest.setMetaParams(metaParams);

//        DmDatasource ds = DmMetaUtils.getDatasourceById(Contants.META_CLASS_NAME, metaRequest);
        DmDatasource ds = DmMetaUtils.getDatasourceByName(Contants.META_CLASS_NAME, metaRequest);
        return (KyligenceConnect) ds.getConnect();
    }

    private DmRequest buildMetaRequest() {
        String authUrl = jobProps.get(Contants.AUTH_SERVER_KEY);
        String dmMetaUrl = jobProps.get(Contants.META_QUERY_URL_KEY);
        DmRequest request = new DmRequest();
        request.setNeedAuth(true);  //
        request.setUrl(dmMetaUrl);
        AuthRequest authRequest = new AuthRequest();
        authRequest.setAuthUrl(authUrl);
        authRequest.setAk(jobProps.get(Contants.AUTH_SERVER_AK));
        authRequest.setSk(jobProps.get(Contants.AUTH_SERVER_SK));
        request.setAuthRequest(authRequest);
        return request;
    }

    private String getBatchScriptParams() {
        //解析json
        JSONObject configJson = parsePluginJson(jobProps.get("plugin_param"));
        //请求数据管理
        KyligenceConnect connect = getDmDataSourceInfo();
        String model = configJson.getString("model");
        String cube = configJson.getString("cube");
        String business_date = configJson.getString("business_date");
        String passwdEncoded = java.util.Base64.getEncoder().encodeToString((connect.getUserName() + ":" + connect.getPassWord()).getBytes());
        return connect.getHost() + " "
                + connect.getPort() + " "
                + passwdEncoded + " "
                + cube + " "
                + business_date; //五个参数
    }

    private String getStreamScriptParams() {
        //解析json
        JSONObject configJson = parsePluginJson(jobProps.get("plugin_param"));
        //请求数据管理
        KyligenceConnect connect = getDmDataSourceInfo();
        String model = configJson.getString("model");
        String cube = configJson.getString("cube");
        String passwdEncoded = java.util.Base64.getEncoder().encodeToString((connect.getUserName() + ":" + connect.getPassWord()).getBytes());
        return connect.getHost() + " "
                + connect.getPort() + " "
                + passwdEncoded + " "  //加密后的字符串
                + cube + " "; //四个参数
    }

}